package com.atguigu.medical.realtime.app.dwd;

import com.atguigu.medical.realtime.app.BaseSQLApp;
import com.atguigu.medical.realtime.common.Constant;
import com.atguigu.medical.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/5/26 14:44
 */
public class DwdTradeConsultationFeePaySuccess extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeConsultationFeePaySuccess().init(
            30002,
            2,
            "DwdTradeConsultationFeePaySuccess"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 因为有 join, 所以需要设置 ttl
        tEnv.getConfig().setIdleStateRetention(Duration.ofHours(5));
        // 1. 读取 ods_medical 表
        readOdsMedical(tEnv, "DwdDiagnosisPrescription");
        // 2. 从 ods_medical 过滤诊费支付成功数据
        Table paymentConsultationFeeTable = tEnv
            .sqlQuery("select " +
                          " after['id'] id,  " +
                          " after['update_time'] payment_time, " +
                          " after['user_id'] user_id, " +
                          " after['consultation_id'] consultation_id, " +
                          " after['payment_amount'] payment_amount," +
                          " ts " +
                          "from ods_medical " +
                          "where source['db']='medical' " +
                          "and source['table']='payment' " +
                          "and op='u' " +
                          "and before['status']='202' " +
                          "and after['status']='203' " +
                          "and after['consultation_id'] is not null ");
        tEnv.createTemporaryView("payment_consultation_fee", paymentConsultationFeeTable);
        
        // 3. 从 ods_medical 过滤出问诊数据
        Table consultationTable = tEnv
            .sqlQuery("select " +
                          " after['id'] id,  " +
                          " after['doctor_id'] doctor_id,  " +
                          " after['user_id'] user_id,  " +
                          " after['patient_id'] patient_id,  " +
                          " ts  " +
                          "from ods_medical " +
                          "where source['db']='medical' " +
                          "and source['table']='consultation' " +
                          "and op='c'"
                          /*"and op='u' " +
                          "and before['status']='202' " +
                          "and after['status']='203' "*/
            );
        tEnv.createTemporaryView("consultation", consultationTable);
        // 4. 2 张表进行 join
        Table result = tEnv.sqlQuery("select " +
                                        " pcf.id, " +
                                        " pcf.payment_time, " +
                                        " pcf.user_id, " +
                                        " pcf.consultation_id, " +
                                        " con.doctor_id," +
                                        " con.patient_id, " +
                                        " pcf.payment_amount, " +
                                        " pcf.ts " +
                                        "from payment_consultation_fee pcf " +
                                        "join consultation con " +
                                        "on pcf.consultation_id=con.id");
        // 5. 最终结果写出到 kafka
        tEnv.executeSql("create table dwd_trade_consultation_fee_pay_success(" +
                            " id string, " +
                            " payment_time string, " +
                            " user_id string, " +
                            " consultation_id string, " +
                            " doctor_id string," +
                            " patient_id string, " +
                            " payment_amount string, " +
                            " ts bigint " +
                            ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CONSULTATION_FEE_PAY_SUCCESS));
        result.executeInsert("dwd_trade_consultation_fee_pay_success");
        
        
    }
}
